package com.tlgen.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.tlgen.Model.RegistryServer;
import com.tlgen.factory.RegistryLocalFactory;
import com.tlgen.service.HeartBeatService;
import com.tlgen.service.RegistryService;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.stream.Collectors;

/*
 *@title HeartBeatServiceImpl
 *@description
 *@author ta151486
 *@version 1.0
 *@create 2023/12/13 15:14
 */
@Service
public class HeartBeatServiceImpl implements HeartBeatService {

    @Autowired
    private RegistryService registryService;

    @Override
    public void sendBeat() {
        List<RegistryServer> registryList = RegistryLocalFactory.getInstance().getRegistryList();
        List<RegistryServer> collect = registryList.stream().peek(x -> x.setDateTime(LocalDateTime.now())).collect(Collectors.toList());
        RegistryLocalFactory.getInstance().setRegistryList(collect);
        for (RegistryServer registryServer : registryList) {
            String serverIP = registryServer.getIp(); // 设置目标服务器的IP地址
            int serverPort = registryServer.getPort(); // 设置目标服务器的端口号
            try (DatagramSocket socket = new DatagramSocket()) {
                InetAddress address = InetAddress.getByName(serverIP);
                String data = "send beat detection ...";
                byte[] buffer = data.getBytes(StandardCharsets.UTF_8); // 要发送的数据内容
                DatagramPacket packet = new DatagramPacket(buffer, buffer.length, address, serverPort);
                socket.send(packet); // 发送心跳包到目标服务器
                socket.close();
                System.out.println("已向 " + serverIP + ":" + serverPort + " 发送心跳包！");
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public void remove() {
        // 获取非健康节点
        List<RegistryServer> registryList = RegistryLocalFactory.getInstance().getRegistryList();
        List<RegistryServer> collect = registryList.stream().filter(x -> {
            long seconds = Duration.between(x.getDateTime(), LocalDateTime.now()).getSeconds();
            return seconds > 6;
        }).collect(Collectors.toList());
        // 设置健康节点
        if (CollectionUtils.isNotEmpty(collect)) {
            System.out.println("非健康节点: " + JSONObject.toJSONString(collect));
            registryList.removeAll(collect);
            System.out.println("剩下健康的节点: " + JSONObject.toJSONString(registryList));
            RegistryLocalFactory.getInstance().setRegistryList(registryList);
            // 将健康节点的服务列表复制到其他节点
            registryService.notifyAllInstance();
        }
    }

    @Override
    public void receive() {
        // 接收 udp 心跳
        try {
            DatagramSocket socket = new DatagramSocket(9995); // 创建监听 UDP 数据的socket，并绑定端口号
            byte[] buffer = new byte[1024]; // 创建缓冲区用于存储接收到的数据
            DatagramPacket packet = new DatagramPacket(buffer, buffer.length); // 创建数据包用于接收UDP数据
            while (true) {
                socket.receive(packet); // 接收 UDP 数据，此方法会阻塞直到接收到数据为止
                String receivedData = new String(packet.getData(), 0, packet.getLength(), StandardCharsets.UTF_8); // 从数据包中获取接收到的数据
                System.out.println("接收回传的 ACK " + packet.getAddress() + ":" + packet.getPort() + ": ");
                System.out.println(receivedData);
                packet.setLength(buffer.length); // 重置数据包的长度，以便下一次接收数据
                if (!receivedData.startsWith("ACK")) {
                    // 构造 ACK 确认消息
                    String ackMessage = "ACK";
                    byte[] ackData = ackMessage.getBytes(StandardCharsets.UTF_8);
                    // 设置 ACK 确认消息的源地址、端口等属性
                    InetAddress sourceAddress = packet.getAddress();
                    int sourcePort = packet.getPort();
                    // 创建 ACK 确认数据包
                    DatagramPacket ackPacket = new DatagramPacket(ackData, ackData.length, sourceAddress, sourcePort);
                    // 发送 ACK 确认消息
                    socket.send(ackPacket);
                    System.out.println("Sent ACK confirmation to sender.");
                } else {
                    // 更新服务列表中健康服务的心跳时间
                    List<RegistryServer> registryList = RegistryLocalFactory.getInstance().getRegistryList();
                    List<RegistryServer> collect = registryList.stream().peek(x -> x.setDateTime(LocalDateTime.now())).collect(Collectors.toList());
                    RegistryLocalFactory.getInstance().setRegistryList(collect);
                }
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

}
